package com.flute.haflute.client;

import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.MalformedURLException;
import java.rmi.Naming;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.Properties;

import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.common.ClusterConfLoader;
import com.flute.haflute.common.ClusterConstants;
import com.flute.haflute.common.ClusterContext;
import com.flute.haflute.jobbox.base.CloudJobConf;
import com.flute.haflute.jobbox.base.JobSubmitter;
import com.flute.haflute.jobbox.common.tools.OSUtil;
import com.flute.haflute.net.messaging.Message;
import com.flute.haflute.net.messaging.MessagingService;
import com.flute.haflute.tools.Utils;

/**
 * 向集群发送工作请求的接口，发送工作请求分为三个阶段：<br>
 * 阶段A、阶段B、阶段C
 * 
 */
public class ClientJobTracer extends IoHandlerAdapter implements JobTracer {

	private static Logger logger = LoggerFactory
			.getLogger(ClientJobTracer.class);

	/** 一项工作的具体配置，所有配置都在这里面 */
	private CloudJobConf jobConf;
	private ClientJobProcess process;
	private boolean responseReceived = false;
	private boolean responseOK = false;
	private String dm = "localhost";
	private int dmPort = 12012;

	/**
	 * 通过该接口提交一项工作请求
	 * 
	 * @param jobConf
	 *            CloudJobConf：任务执行的配置文件
	 * @throws Exception
	 */
	public void submitJob(CloudJobConf jobConf) throws Exception {
		ClusterConfLoader.loadGlobalContext();
		dm = ClusterContext.getDM();
		dmPort = ClusterContext.getHeartBeatPort();
		submitJob(jobConf, dm, dmPort);
	}

	/**
	 * 根据工作请求向集群发送消息——阶段A
	 * 
	 * @param jobConf
	 *            任务的配置文件，类：CloudJobConf
	 * @param clusterDm
	 *            指定集群主节点IP或者名字
	 * @param port
	 *            主节点的端口
	 */
	public void submitJob(CloudJobConf jobConf, String clusterDm, int port)
			throws Exception {
		Object jc = jobConf.getJobContext();
		if (jc instanceof Properties) {
			jobConf.setJobDescription(getDescription(getFileArray((Properties) jc)));
		}
		this.jobConf = jobConf;
		InetAddress myIp = ClusterContext.getLocalAddress();
		if (myIp == null)
			myIp = OSUtil.getLocalHostAddress(false, null);
		Message message = new Message(myIp, Message.MessageType.JOB_NEW.name(),
				Message.Verb.JOB_NEW, Utils.serialize(jobConf));
		MessagingService.getInstance().send2Node(
				new InetSocketAddress(clusterDm, port), message, this);
	}

	/**
	 * 向工作控制中心节点提交工作请求——阶段C
	 * 
	 * @param jobConf
	 *            任务的配置文件，类：CloudJobConf
	 * @param masterIp
	 *            需要执行任务的主节点IP或者名字
	 * @param rmiPort
	 *            主节点的端口
	 */
	public void runJob(CloudJobConf jobConf, String masterIp, int rmiPort)
			throws MalformedURLException, RemoteException, NotBoundException {
		System.out.println("please wait, I am submitting 1 cloud job to "
				+ masterIp + ":" + rmiPort);

		// 获取主节点任务提交的接口，并提交任务
		for (int i = 0; i < 3; i++) {
			try {
				JobSubmitter jobSubmitter = (JobSubmitter) Naming
						.lookup("rmi://" + masterIp + ":" + rmiPort
								+ "/JOB_SUBMITTER");
				Object jc = jobConf.getJobContext();
				if (jc instanceof Properties) {
					String[] fileNames = getFileArray((Properties) jc);
					if (fileNames != null && fileNames.length > 0) {
						TransferredFiles fileContents = getTransferFiles(fileNames);
						// 将文件传送到执行作业的节点上
						jobSubmitter.transferFiles(fileContents);
					}
				}

				process = new ClientJobProcess();

				jobSubmitter.submit(jobConf, process);

				System.out
						.println("Job submitted OK, please wait for Job to finish");
				responseOK = true;
				break;
			} catch (Exception e) {
				e.printStackTrace();
				logger.error("", e);
				if (i < 2) {
					long waittime = 1000L * (i * 3 + 1);
					System.out.println("submit failed, will try in  "
							+ waittime + " seconds");
					await(waittime);
				} else {
					responseOK = false;
					System.err
							.println("can NOT submit Job, please check the Cluster status");
				}
			}
		}
	}

	private String[] getFileArray(Properties props) {
		// Check if have files to send
		ArrayList<String> fileList = new ArrayList<String>();
		for (Object obj : props.keySet()) {
			String key = (String) obj;
			if (key.startsWith(Constants.FILEPREFIX)) {
				fileList.add(props.getProperty(key));
			}
		}
		return fileList.toArray(new String[fileList.size()]);
	}

	private void await(long waittime) {
		try {
			Thread.sleep(waittime);
		} catch (InterruptedException e1) {
		}

	}

	/**
	 * 接收集群是否可以执行工作请求消息 如果可以，则根据返回的该工作控制节点，向该节点发送工作配置信息 ——阶段B 否则报错。
	 * 
	 * @see org.apache.mina.core.service.IoHandlerAdapter#messageReceived(org.apache.mina.core.session.IoSession,
	 *      java.lang.Object)
	 */
	@Override
	public void messageReceived(IoSession session, Object message)
			throws Exception {
		if (message instanceof Message) {
			Message realMessage = (Message) message;
			if (realMessage.getMessageType().equals(
					Message.MessageType.JOB_NEW.name())) {
				// 从消息里面取得该工作分配的控制节点
				String master = new String(
						realMessage.getHeader(ClusterConstants.JOB_ROLL.MASTER
								.name()));
				// 向该控制节点发出工作请求
				try {
					runJob(jobConf, master, ClusterContext.getRMIPort());
				} catch (Exception e) {
					e.printStackTrace();
					logger.error("", e);
				}
			} else {
				System.out
						.println("Cluster is NOT ready for new job to execute, please submit job later");
				responseOK = false;
			}

			if (responseOK)
				Utils.sendMessage(session, getSubmitResultMessage(true));
			else
				Utils.sendMessage(session, getSubmitResultMessage(false));

			responseReceived = true;
			session.close(true);
		}
	};

	private TransferredFiles getTransferFiles(String[] fileNames)
			throws IOException {
		TransferredFiles needFiles = new TransferredFiles();
		for (String file : fileNames) {
			try {
				needFiles.addOneFile(file);
			} catch (IOException e) {
				System.out.println("one File add failed: " + file);
				throw e;
			}
		} // for
		return needFiles;
	}

	private void getPartOfFilePath(final String originFilePath,
			final int level, StringBuilder result) {
		if (level <= 0) {
			return;
		}
		int i = originFilePath.lastIndexOf(File.separator);
		if (i > 0) {
			getPartOfFilePath(originFilePath.substring(0, i), level - 1, result);
		}
		result.append("/").append(originFilePath.substring(i + 1));
	}

	private String getDescription(String[] fileNames) {
		StringBuilder sb = new StringBuilder();
		for (String file : fileNames) {
			if (sb.length() > 0)
				sb.append("_");
			getPartOfFilePath(file, 2, sb);
		}
		return sb.toString();
	}

	private Message getSubmitResultMessage(boolean success) throws IOException {
		String messageType = Message.MessageType.JOB_SCHEDULED.name();
		if (!success) {
			messageType = Message.MessageType.JOB_ENDED.name();
		}
		Message message = new Message(ClusterContext.getLocalAddress(),
				messageType, Message.Verb.JOB_NEW, Utils.serialize(jobConf));
		return message;
	}

	/**
	 * 如果提交任务成功，返回true。不成功的原因是集群资源暂时不够
	 * 
	 * @return 任务是否成功
	 */
	public boolean awaitSubmitSuccessUninterruptly() {
		while (!responseReceived) {
			await(1000);
		}
		return responseOK;
	}

	/**
	 * @return the JOB是否完成
	 */
	public boolean isJobFinished() {
		if (process == null) {
			return false;
		}
		return process.isJobFinished();
	}
}
